/*
* Change Logs:
* Date           Author       Notes
* 2020-10-16     chenbin      mqtt client base on paho
*/
#include <string.h>
#include <stdint.h>

#include "small_mqtt_client.h"

int mqtt_printf(mqtt_client_t *client,int level,const char *fmt, ...)
{
	static char log_buf[256];
	rt_enter_critical();
	//if(level <= client->debug_level)
	{
		va_list args;
		va_start(args, fmt);
		rt_vsnprintf(log_buf, 256, fmt, args);
		va_end(args);
		rt_kputs("[MQTT LOG] ");
		rt_kputs(log_buf);
	}
	rt_exit_critical();
	return MQTT_OK;
}
/*
*可自行实现的函数,根据不同平台不同设备实现TCP网络通信
*可用方式 
*1.socket 接口需要平台支持
*2.rtt device 可以对接各种硬件平台网络透传模块，实现上层mqtt通信
*/
//////////////////////////////////////////////////////////////////
int mqtt_net_connect(mqtt_client_t *client)
{
	//return MQTT_OK;
	if((client->port)&&(client->port->net_connect))
	{
		return client->port->net_connect(client);
	}
	return MQTT_OK;
}

int mqtt_net_disconnect(mqtt_client_t *client)
{
	//return MQTT_OK;
	if((client->port)&&(client->port->net_disconnect))
	{
		return client->port->net_disconnect(client);
	}
	return MQTT_OK;
}
/*
*返回>0数据长度
*返回=0没有接收到数据
*返回-1为一般错误,网络且正常
*返回-2为等待数据超时,网络且正常
*返回-3为网络断开,网络异常
 */
int mqtt_net_recv(mqtt_client_t *client, unsigned char *buf, int len, int timeout_ms)
{
	//return MQTT_OK;
	if((client->port)&&(client->port->net_recv))
	{
		return client->port->net_recv(client,buf,len,timeout_ms);
	}
	return MQTT_OK;
}

int mqtt_net_send(mqtt_client_t *client, unsigned char *buf, int len)
{
	//return MQTT_OK;
	if((client->port)&&(client->port->net_send))
	{
		return client->port->net_send(client,buf,len);
	}
	return MQTT_OK;
}

int mqtt_lock_take(mqtt_client_t *client, int timeout_ms)
{
	rt_mutex_t mutex = client->write_lock;
	if(timeout_ms<RT_WAITING_FOREVER)
		timeout_ms = RT_WAITING_FOREVER;
	return rt_mutex_take(mutex,timeout_ms);
}

int mqtt_lock_release(mqtt_client_t *client)
{
	rt_mutex_t mutex = client->write_lock;
	return rt_mutex_release(mutex);
}

void mqtt_delayms(int ms)
{
	rt_thread_mdelay(ms);
}
/*mqtt tick*/
uint32_t mqtt_tick_get(void)
{
	return rt_tick_get();
}
/*参数检查*/
int mqtt_config_check(mqtt_client_t *client)
{

    if(client->reconnect_interval == 0)
    {
      client->reconnect_interval = 10;
    }
    if(client->write_timeout == 0)
    {
      client->write_timeout = 500;
    }
    if(client->write_size == 0)
    {
        client->write_size = 1024;
    }
    if(client->write_buff == NULL)
    {
        client->write_size = RT_ALIGN(client->write_size,4);
        client->write_buff = rt_malloc(client->write_size);
    }
    if(client->read_timeout == 0)
    {
      client->read_timeout = 500;
    }
    if(client->read_size == 0)
    {
        client->read_size = 1024;
    }
    if(client->read_buff == NULL)
    {
        client->read_size = RT_ALIGN(client->read_size,4);
        client->read_buff = rt_malloc(client->read_size);
    }
		
    client->_ping_keepalive = client->condata.keepAliveInterval;
		client->_sub_max =  (sizeof(client->message_handle)/sizeof(struct message_handlers));
		
		if((client->read_size==0)||(client->read_buff == NULL))
		{
			mqtt_debug(client,"read size or buff is NULL\n");
		}
		if((client->write_size==0)||(client->write_buff == NULL))
		{
			mqtt_debug(client,"write size or buff is NULL\n");
		}
		if(client->write_lock == NULL)
		{
			client->write_lock = rt_mutex_create("mqttlock",RT_IPC_FLAG_FIFO);
		}
    return MQTT_OK;
}

//////////////////////////////////////////////////////////////////

static int decode_packet(mqtt_client_t *client, int *value, int timeout_ms)
{
  unsigned char i;
  int multiplier = 1;
  int len = 0;
  const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4;

  *value = 0;
  do
  {
    int rc = MQTTPACKET_READ_ERROR;

    if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES)
    {
      rc = MQTTPACKET_READ_ERROR; /* bad data */
      goto exit;
    }
    rc = mqtt_net_recv(client, &i, 1, timeout_ms);
    if (rc != 1)
      goto exit;
    *value += (i & 127) * multiplier;
    multiplier *= 128;
  }
  while ((i & 128) != 0);
exit:
  return len;
}

static int mqtt_readpacket(mqtt_client_t *client, int waittime_ms)
{
  int rc = MQTT_FAIL;
  MQTTHeader header = {0};
  int len = 0;
  int rem_len = 0;

	if(waittime_ms>0)
	{
		rc = mqtt_net_recv(client, client->read_buff, 1,waittime_ms);
	}else
	{
		rc = mqtt_net_recv(client, client->read_buff, 1,client->read_timeout);
	}
  /* 1. read the header byte.  This has the packet type in it */
  if (rc != 1)
    return rc;

  len = 1;
  /* 2. read the remaining length.  This is variable in itself */
  decode_packet(client, &rem_len, 50);
  len += MQTTPacket_encode(client->read_buff + 1, rem_len); /* put the original remaining length back into the buffer */
  if (rem_len > (client->read_size - len))
  {
		return MQTT_FAIL;
  }
  /* 3. read the rest of the buffer using a callback to supply the rest of the data */
  if (rem_len > 0 && (mqtt_net_recv(client, client->read_buff + len, rem_len, 300) != rem_len))
  {
		return MQTT_FAIL;
  }

  header.byte = client->read_buff[0];
  rc = header.bits.type;
  return rc;
}

static int get_next_packetId(mqtt_client_t *client)
{
  return client->_next_packetid = (client->_next_packetid == MAX_PACKET_ID) ? 1 : client->_next_packetid + 1;
}

static char is_topic_matched(char *topicFilter, MQTTString *topic_name)
{
  char *curf = topicFilter;
  char *curn = topic_name->lenstring.data;
  char *curn_end = curn + topic_name->lenstring.len;

  while (*curf && curn < curn_end)
  {
    if (*curn == '/' && *curf != '/')
      break;
    if (*curf != '+' && *curf != '#' && *curf != *curn)
      break;
    if (*curf == '+')
    {
      // skip until we meet the next separator, or end of string
      char *nextpos = curn + 1;
      while (nextpos < curn_end && *nextpos != '/')
        nextpos = ++curn + 1;
    }
    else if (*curf == '#')
      curn = curn_end - 1;    // skip until end of string
    curf++;
    curn++;
  };

  return (curn == curn_end) && (*curf == '\0');
}

static int deliver_message(mqtt_client_t *client, mqtt_message_t *message)
{
  int i;
  int rc = MQTT_FAIL;
  // we have to find the right message handler - indexed by topic
  for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
  {
    if (client->message_handle[i].topicName != 0 &&
            (MQTTPacket_equals( &(message->topic), (char *)client->message_handle[i].topicName) ||
            is_topic_matched((char *)client->message_handle[i].topicName, &(message->topic) )))
    {
      if (client->message_handle[i].callback != NULL)
      {
        client->message_handle[i].callback(client, message);
        rc = MQTT_OK;
      }
    }
  }
  if (rc == MQTT_FAIL && client->message_handle_default != NULL)
  {
    client->message_handle_default(client, message);
    rc = MQTT_OK;
  }
  return rc;
}

static int _small_mqtt_connect(mqtt_client_t *client)
{
  int rc = MQTT_FAIL, len;
  MQTTPacket_connectData *options = &client->condata;
	
	MQTT_CHECK_NET(client);  //check network
	
	if(mqtt_lock_take(client,-1)==MQTT_OK)  //take lock
	{
		len = MQTTSerialize_connect(client->write_buff, client->write_size, options);
		if(len > 0)
		{
				rc = mqtt_net_send(client, client->write_buff, len);
		}
		mqtt_lock_release(client);  //release lock
	}
  return rc;
}

static int _small_mqtt_disconnect(mqtt_client_t *client)
{
  int rc = MQTT_FAIL;
  int len = 0;
	
	MQTT_CHECK_NET(client);  //check network
	
	if(mqtt_lock_take(client,-1)==MQTT_OK)  //take lock
	{
		len = MQTTSerialize_disconnect(client->write_buff, client->write_size);
		if (len > 0)
		{
			rc = mqtt_net_send(client, client->write_buff, len);
		}
		mqtt_lock_release(client);  //release lock
	}
  return rc;
}

static int _small_mqtt_subscribe(mqtt_client_t *client, const char *topicName, enum QoS qos)
{
  int rc = MQTT_FAIL;
  int len = 0;
  int qos_sub = qos;
  MQTTString topic = MQTTString_initializer;

	MQTT_CHECK_NET(client);  //check network
	
	if(mqtt_lock_take(client,-1)==MQTT_OK)  //take lock
	{
		topic.cstring = (char *)topicName;
		len = MQTTSerialize_subscribe(client->write_buff, client->write_size, 0, get_next_packetId(client), 1, &topic, &qos_sub);
		if(len > 0)
		{
				rc = mqtt_net_send(client, client->write_buff, len);
		}
		mqtt_lock_release(client);  //release lock
	}
  return rc;
}

static int _small_mqtt_unsubscribe(mqtt_client_t *client, const char *topicName)
{
  int rc = MQTT_FAIL;
  int len = 0;
  MQTTString topic = MQTTString_initializer;

	MQTT_CHECK_NET(client);  //check network
	
	if(mqtt_lock_take(client,-1)==MQTT_OK)  //take lock
	{
		topic.cstring = (char *)topicName;
		len = MQTTSerialize_unsubscribe(client->write_buff, client->write_size, 0, get_next_packetId(client), 1, &topic);
		if(len > 0)
		{
				rc = mqtt_net_send(client, client->write_buff, len);
		}
		mqtt_lock_release(client);  //release lock
	}
  return rc;
}

static int _small_mqtt_publish(mqtt_client_t *client, enum QoS qos, const char *topicName, void *payload, size_t length)
{
  int rc = MQTT_FAIL;
	int len;
	uint16_t message_id;
	MQTTString mqtt_topic = MQTTString_initializer;

	MQTT_CHECK_NET(client);  //check network
	
	if(mqtt_lock_take(client,-1)==MQTT_OK)  //take lock
	{
		mqtt_topic.cstring = (char *)topicName;
		if (qos == QOS1 || qos == QOS2)
		{
			message_id = get_next_packetId(client);
		}else
		{
			message_id = 0;
		}
		len = MQTTSerialize_publish(client->write_buff, client->write_size, 0, qos, 0, message_id,mqtt_topic, payload, length);
		if(len>0)
		{
			rc = mqtt_net_send(client, client->write_buff, len);
			if(rc < MQTT_OK)
			{
				mqtt_debug(client,"net send error(%d)\n", rc);
				client->status_net = MQTT_NET_DIS;
			}else
			{
				client->_count_client_publish++;  //client publish count
				rc = MQTT_OK;
			}
		}
		mqtt_lock_release(client);  //release lock
	}
	return rc;
}

static int _small_mqtt_ping_server(mqtt_client_t *client)
{
  int rc = MQTT_FAIL;
  int len = 0;
	
	MQTT_CHECK_NET(client);  //check network
	
	if(mqtt_lock_take(client,-1)==MQTT_OK)  //take lock
	{
		len = MQTTSerialize_pingreq(client->write_buff, client->write_size);
		if(len>0)
		{
			rc = mqtt_net_send(client, client->write_buff, len);
			if(rc < MQTT_OK)
			{
				mqtt_debug(client,"net send error(%d)\n", rc);
				client->status_net = MQTT_NET_DIS;
			}else
			{
				client->_count_ping++;  //client ping count
				rc = MQTT_OK;
			}
		}
		mqtt_lock_release(client);  //release lock
	}
  return rc;
}

static int _small_mqtt_handle_data(mqtt_client_t *client, int waittime_ms)
{
    int len = 0;
    int ret = 0;
    int rc = MQTT_OK;;
    // read the socket, see what work is due
    int packet_type = mqtt_readpacket(client,waittime_ms);
    if (packet_type < MQTT_OK)
    {
				if(packet_type == MQTT_TIMEOUT)
				{
					packet_type = MQTT_OK;  //time get tick
				}
        if(packet_type == MQTT_RESTART)
        {
            if(client->status_handle)
                client->status_handle(client,MQTT_DISCONNECT);
        }
        return packet_type;
    }
    switch (packet_type)
    {
    case CONNACK:
         rc = MQTT_OK;
      break;
    case PUBACK:
      {
        uint16_t mypacketid;
        uint8_t dup, type;
        ret = MQTTDeserialize_ack(&type, &dup, &mypacketid, client->read_buff, client->read_size);
        if (ret == 1)
        {
            rc = MQTT_OK;
        }
      }
      break;
    case PUBCOMP:
      {
        uint16_t mypacketid;
        uint8_t dup, type;
        ret = MQTTDeserialize_ack(&type, &dup, &mypacketid, client->read_buff, client->read_size);
        if (ret == 1)
        {
            rc = MQTT_OK;
        }
      }
      break;
    case SUBACK:
      {
        int count = 0, grantedQoS = -1;
        unsigned short mypacketid;
        ret = MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, client->read_buff, client->read_size);
        if (ret == 1)
        {
            rc = MQTT_OK;
        }
      }
      break;
    case UNSUBACK:
      {
        unsigned short mypacketid;
        ret = MQTTDeserialize_unsuback(&mypacketid, client->read_buff, client->read_size);
        if (ret == 1)
        {
            rc = MQTT_OK;
        }
      }
      break;
    case PUBLISH:
      {
        //MQTTString topic_name;
        mqtt_message_t msg;
        int intQoS;
        ret = MQTTDeserialize_publish(&msg.dup, &intQoS,&msg.retained, &msg.id, 
					&(msg.topic),(unsigned char **)&msg.payload,(int *)&msg.payloadlen,
					client->read_buff, client->read_size);
        if(ret == 1)
        {
						client->_count_server_publish++;  //server publish count
						
            msg.qos = (enum QoS)intQoS;
            if (msg.qos != QOS0)
            {
							if(mqtt_lock_take(client,-1)==MQTT_OK)  //take lock
							{
								if (msg.qos == QOS1)
									len = MQTTSerialize_ack(client->write_buff, client->write_size, PUBACK, 0, msg.id);
								else if (msg.qos == QOS2)
									len = MQTTSerialize_ack(client->write_buff, client->write_size, PUBREC, 0, msg.id);
								if (len)
								{
									rc = mqtt_net_send(client, client->write_buff, len);
								}
								mqtt_lock_release(client);  //release lock
							}
            }
            deliver_message(client,&msg);
            rc = MQTT_OK;
        }else {
            rc = MQTT_FAIL;
        }
      }
      break;
    case PUBREC:
    case PUBREL:
      {
        unsigned short mypacketid;
        unsigned char dup, type;
        ret = MQTTDeserialize_ack(&type, &dup, &mypacketid, client->read_buff, client->read_size);
        if (ret != 1)
        {
					if(mqtt_lock_take(client,-1)==MQTT_OK)  //take lock
					{
						len = MQTTSerialize_ack(client->write_buff, client->write_size,(packet_type == PUBREC) ? PUBREL : PUBCOMP, 0, mypacketid);
						if(len)
						{
							ret = mqtt_net_send(client, client->write_buff, len);
						}
						mqtt_lock_release(client);  //release lock
					}
					rc = MQTT_OK;	
        }
      }
      break;
    case PINGRESP:
			client->_count_pingresp++; // add pingresp count
      break;
    }
    if(rc == MQTT_OK)
    {
        if(client->status_handle)
            client->status_handle(client,packet_type);
    }
    return rc;
}

int _small_mqtt_handle_status(mqtt_client_t *client)
{
	int rc = MQTT_OK;
	uint32_t tick = 0;
	switch(client->status_run)
	{
		case MQTT_RUN_CON:
		{
			rc = _small_mqtt_connect(client);
			if (rc < MQTT_OK)
			{
					mqtt_debug(client,"connect error(%d)\n", rc);
					return MQTT_RESTART;
			}
			rc = MQTT_OK;
			client->status_run = MQTT_RUN_SUB;
		}break;
		case MQTT_RUN_SUB:
		{
			const char *topic = NULL;
			enum QoS qos;
			do{
				topic = client->message_handle[client->_sub_index].topicName;
				qos = client->message_handle[client->_sub_index].qos;
				client->_sub_index++;
			}while((topic == NULL)&&(client->_sub_index < client->_sub_max));
			
			if(topic != NULL)
			{
					rc = _small_mqtt_subscribe(client, topic, qos);
					client->_sub_used++;
					mqtt_debug(client,"subscribe #%d %s\n", client->_sub_index, topic);
					if (rc < MQTT_OK)
					{
							mqtt_debug(client,"subscribe error(%d)\n", rc);
							return MQTT_RESTART;
					}
			}else
			{
				mqtt_debug(client,"subscribe count %d  used %d\n", client->_sub_max,client->_sub_used);
				client->_sub_index=0;
				client->_sub_used = 0;
				client->_ping_count = 0;
				client->_ping_tick = 0;
				
				client->_count_ping = 0;			  //ping count reset
				client->_count_pingresp = 0;  //ping count reset
				
				client->status_run = MQTT_RUN_IDLE;
			}
			rc = MQTT_OK;
		}break;
		case MQTT_RUN_DIS:
		{
			//send mqtt disconnect pack
			rc = _small_mqtt_disconnect(client);
			if (rc < MQTT_OK)
			{
					mqtt_debug(client,"disconnect error(%d)\n", rc);
					return MQTT_RESTART;
			}
			client->status_run = MQTT_RUN_IDLE;
		}break;
		case MQTT_RUN_IDLE:
		default:
		{
			tick = (mqtt_tick_get() - client->_ping_tick); //tick diff
			client->_ping_tick = mqtt_tick_get();	//get now tick
			if(client->_ping_count > tick)
			{
				client->_ping_count -= tick;
			}else
			{
				//check ping count > pingresp count
				if((client->_count_ping - client->_count_pingresp ) >= 2)
				{
					mqtt_debug(client,"ping count:%d  pingresp count:%d error\n", client->_count_ping, client->_count_pingresp );
					
					client->status_run = MQTT_RUN_DIS;
					return MQTT_RESTART;
				}
				//send ping 
				client->_ping_count = client->_ping_keepalive * 1000;
				rc = _small_mqtt_ping_server(client);
				if (rc < MQTT_OK)
				{
						mqtt_debug(client,"ping error(%d)\n", rc);
						return MQTT_RESTART;
				}
			}
			
			rc = MQTT_OK;
		}break;
	}
	return rc;
}

void small_mqtt_thread(void *param)
{
	mqtt_client_t *client = (mqtt_client_t *)param;
	int rc = 0;
	//mqtt config check
	mqtt_config_check(client);
	client->status_run = MQTT_RUN_INIT;
	client->status_net = MQTT_NET_DIS;
	client->_sub_index = 0;
	
	mqtt_debug(client,"client start!\n");
	//main loop
	while(client->status_run != MQTT_RUN_EXIT)
	{
		client->status_run = MQTT_RUN_INIT;
		client->status_net = MQTT_NET_DIS;
		client->_sub_index = 0;
		
		rc = mqtt_net_connect(client);
		if(rc < MQTT_OK)
		{
			mqtt_debug(client,"net connect error(%d)\n", rc);
			client->status_net = MQTT_NET_DIS;
			client->status_run = MQTT_RUN_INIT;
		}else
		{
			client->status_net = MQTT_NET_CON;
			client->status_run = MQTT_RUN_CON;
		}
		//mqtt client main loop
		while((client->status_net == MQTT_NET_CON)&&
		(client->status_run > MQTT_RUN_INIT))
		{
			rc = _small_mqtt_handle_status(client);
			if(rc < MQTT_OK) //ignore MQTT_TIMEOUT MQTT_FAIL MQTT_OK 
			{
				mqtt_debug(client,"status error(%d)\n", rc);
				client->status_net = MQTT_NET_DIS;
			}
			rc =  _small_mqtt_handle_data(client,rc);  //blocking wait data
			if(rc < MQTT_OK) //ignore MQTT_TIMEOUT MQTT_FAIL MQTT_OK 
			{
				mqtt_debug(client,"data error(%d)\n", rc);
				client->status_net = MQTT_NET_DIS;
			}
		}
		//network disconnect
		mqtt_net_disconnect(client);
		
		client->_count_reconnect++;
		//delay
		mqtt_delayms(client->reconnect_interval*1000);
		mqtt_debug(client,"client restart!\n");
	}
}

int small_mqtt_subscribe(mqtt_client_t *client, enum QoS qos,char *topic, mqtt_message_cb callback)
{
    int rc = MQTT_FAIL;
    int i;
    int qos_sub = qos;

    if((client == NULL) || (topic == NULL))
		{
        return rc;
    }

    for (i = 0; i < MAX_MESSAGE_HANDLERS ; ++i)
    {
        if (client->message_handle[i].topicName &&
            rt_strncmp(client->message_handle[i].topicName, topic, rt_strlen(topic)) == 0)
        {
          mqtt_debug(client,"topic(%s) is already subscribed.\n", topic);
          return MQTT_OK;
        }
    }

    for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
    {
        if (client->message_handle[i].topicName == NULL)
        {
					client->message_handle[i].qos = qos;
					client->message_handle[i].topicName = topic;
					if (callback)
					{
							client->message_handle[i].callback = callback;
					}
					if(_small_mqtt_subscribe(client,topic,(enum QoS)qos_sub) > MQTT_OK)
					{
						mqtt_debug_info(client,"subscribe #%d %s OK!\n",i,topic);
					}else
					{
						mqtt_debug_info(client,"subscribe #%d %s add to handle\n",i,topic);
					}
					rc = MQTT_OK;
          break;;
        }
    }
    /* check subscribe numble support */
    if (i >= MAX_MESSAGE_HANDLERS) {
        mqtt_debug(client,"subscribe MAX_MESSAGE_HANDLERS size(%d) is not enough!\n", MAX_MESSAGE_HANDLERS);
        rc = MQTT_FAIL;
    }
    return rc;
}

int small_mqtt_unsubscribe(mqtt_client_t *client,char *topic)
{
	int rc = MQTT_FAIL;
	int i;

	MQTT_CHECK_NET(client);

	if(topic == NULL)
	{
			return rc;
	}

	for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
	{
			if (client->message_handle[i].topicName == RT_NULL ||
					rt_strncmp(client->message_handle[i].topicName, topic, rt_strlen(topic)) != 0)
			{
				continue;
			}
			rc = _small_mqtt_unsubscribe(client,topic);
			mqtt_debug_info(client,"unsubscribe #%d %s  rc:%d\n",i, client->message_handle[i].topicName ,rc);
	}

	/* check subscribe topic */
	if (i >= MAX_MESSAGE_HANDLERS)
	{
			mqtt_debug(client,"unsubscribe topic(%s) is not exist!\n", topic);
			rc = MQTT_FAIL;
	}
	return rc;
}

int small_mqtt_publish(mqtt_client_t *client, enum QoS qos, char *topic, void *payload, size_t length)
{
	MQTT_CHECK_NET(client);
	
	if( (topic == NULL) || (length == 0))
	{
		return MQTT_FAIL;
	}
	return _small_mqtt_publish(client,qos,topic,payload,length);
}

int small_mqtt_will(mqtt_client_t *client,unsigned char willFlag,unsigned char retained, enum QoS qos,char *topic,char *msg)
{
	int rc = MQTT_FAIL;
	
	if((client == NULL) || (topic == NULL))
	{
		return rc;
	}
	return rc;
}

int small_mqtt_reconnect(mqtt_client_t *client)
{
	int rc = MQTT_FAIL;
	if(client != NULL)
	{
		mqtt_debug(client,"reconnect\n");
		client->status_net = MQTT_NET_DIS;
		
		rc = MQTT_OK;
	}
	return rc;
}


#include "MQTTConnect.h"
static int _mqtt_init(mqtt_client_t *client,void *port)
{
	MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;
	rt_memcpy(&(client->condata),&condata,sizeof(MQTTPacket_connectData));
	
	mqtt_client_port_t *mqtt_port = port;
	
	if((mqtt_port)&&((mqtt_port->magic == MQTT_PORT_DEVICE)||(mqtt_port->magic == MQTT_PORT_SOCKET)))
	{
		client->port = mqtt_port;
	}
	
	client->socket_fd = -1; 
	return 0;
}

int small_mqtt_control(mqtt_client_t *client, int cmd, void *arg)
{
  if((client==NULL)||(arg==NULL))
	{
    return MQTT_FAIL;
  }
  switch(cmd)
  {
  case MQTT_CLIENT_INIT:
		_mqtt_init(client,arg);
    break;
  case MQTT_SET_URI:
		client->chost = arg;
    break;
  case MQTT_SET_HOST:
		client->chost = arg;
    break;
  case MQTT_SET_PORT:
		client->cport = arg;
    break;
	//read
  case MQTT_SET_READ_TIMEOUT_MS:
      client->read_timeout = *(int *)arg;
    break;
  case MQTT_SET_READ_BUFF_SIZE:
      client->read_size = *(int *)arg;
    break;
  case MQTT_SET_READ_BUFF:
      client->read_buff = arg;
    break;
	//write
  case MQTT_SET_WRITE_TIMEOUT_MS:
      client->write_timeout = *(int *)arg;
    break;
  case MQTT_SET_WRITE_BUFF_SIZE:
      client->write_size = *(int *)arg;
    break;
  case MQTT_SET_WRITE_BUFF:
      client->write_buff = arg;
    break;
	//mqtt
  case MQTT_SET_CLIENTID:
		client->condata.clientID.cstring = arg;
    break;
  case MQTT_SET_USERNAME:
		client->condata.username.cstring = arg;
    break;
  case MQTT_SET_PASSWORD:
		client->condata.password.cstring = arg;
    break;
  case MQTT_SET_KEEPALIVE :
    client->condata.keepAliveInterval = *(int *)arg;
    break;
  case MQTT_SET_CLEANSESSION:
    client->condata.cleansession = *(int *)arg;
    break;
  case MQTT_SET_RECONNECT:
    client->reconnect_interval = *(int *)arg;
    break;
  case MQTT_CB_STATUS:
    client->status_handle = (mqtt_status_cb)arg;
    break;
  case MQTT_CB_MESSAGE:
    client->message_handle_default = (mqtt_message_cb)arg;
    break;
  default:
    mqtt_debug(client,"cmd(%d) error.\n", cmd);
    break;
  }
  return MQTT_OK;
}
